"""
pymongo 二次封装
"""

import math

from pymongo import (
    DeleteMany,
    DeleteOne,
    InsertOne,
    ReplaceOne,
    UpdateMany,
    UpdateOne,
)
from pymongo.collection import Collection
from pymongo.cursor import Cursor
from pymongo.results import (
    BulkWriteResult,
    DeleteResult,
    InsertManyResult,
    InsertOneResult,
    UpdateResult,
)

from db_base.mongodb_db.client import mongo_db
from my_typing import *


class empty:  # noqa
    """
    该类用于表示没有为给定的输入或输出值提供数据
    它是必需的，因为 `None` 可能是一个有效的输入或输出值
    """

    pass


class CollectionSimpleUtils:
    """ 集合简单的工具类 """

    projection: DictStrInt = {
        "_id": 0,
    }
    """ 字段展示控制（默认不展示 `_id`） """

    deleted_mark: DictStrInt = {
        "is_deleted": 1,
    }
    """ 数据删除标记 """

    undeleted_mark: DictStrInt = {
        "is_deleted": 0,
    }
    """ 数据未删除标记 """

    undeleted_query: dict[str, DictStrInt] = {
        "is_deleted": {
            "$ne": 1,
        },
    }
    """ 数据未删除查询条件 """

    @staticmethod
    def get_time_dict(*args: str) -> DictStrStr:
        """
        获取时间字段的字典

        Args:
            args: 字段

        Returns:
            时间字段的字典

        Examples:
            >>> # 不传任何字段，返回空字典
            >>> CollectionSimpleUtils.get_time_dict()
            {}

            >>> # 传入 `update_time`，返回只有 1 个元素的字典
            >>> CollectionSimpleUtils.get_time_dict("update_time")
            {
                "update_time": "2023-05-10 15:31:23",
            }

            >>> # 传入 `update_time`、`update_time`，返回有 2 个元素的字典
            >>> CollectionSimpleUtils.get_time_dict("create_time", "update_time")
            {
                "create_time": "2023-05-10 15:31:23",
                "update_time": "2023-05-10 15:31:23",
            }
        """

        if not args:
            return {}

        now_datetime_str: str = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        """ 当前日期时间（格式: %Y-%m-%d %H:%M:%S） """

        return {
            time_field: now_datetime_str
            for time_field in args
        }

    @staticmethod
    def get_deleted_mark_dict(filter_all: bool) -> dict:
        """
        获取删除标记的字典

        Args:
            filter_all: 是否查询全部数据

        Returns:
            删除标记的字典

        Examples:
            >>> # `filter_all` 为 False，获取未删除的数据，即：`is_deleted` 不等于 1 的数据（兼容没有 `is_deleted` 的字段）
            >>> CollectionSimpleUtils.get_deleted_mark_dict(
            ...     filter_all=False,
            ... )
            {
                "is_deleted": {
                    "$ne": 1,
                },
            }

            >>> # `filter_all` 为 True，获取全部数据
            >>> CollectionSimpleUtils.get_deleted_mark_dict(
            ...     filter_all=True,
            ... )
            {}
        """

        if filter_all:
            return {}

        return CollectionSimpleUtils.undeleted_query.copy()

    @staticmethod
    def _get_filter(
        filter: DictOrNone = None,  # noqa: BuiltinShadowName
        filter_all: bool = False,
    ) -> dict:
        """
        获取查询条件

        Args:
            filter: 过滤条件
            filter_all: 是否查询全部数据

        Returns:
            查询条件

        Examples:
            >>> # `filter` 为 None
            >>> CollectionSimpleUtils._get_filter()
            {
                "is_deleted": {
                    "$ne": 1,
                },
            }

            >>> # `filter` 为 None，`filter_all` 为 True
            >>> CollectionSimpleUtils._get_filter(filter_all=True)
            {}

            >>> # `filter` 不为 None，`filter_all` 为 False
            >>> CollectionSimpleUtils._get_filter(
            ...     filter={
            ...         "name": "张三",
            ...         "age": 18,
            ...     },
            ... )
            {
                "name": "张三",
                "age": 18,
                "is_deleted": {
                    "$ne": 1,
                },
            }
        """

        _filter: dict = {
            **(
                {}
                if filter is None
                else filter
            ),
            **CollectionSimpleUtils.get_deleted_mark_dict(filter_all)
        }

        return _filter

    @staticmethod
    def _get_projection(projection: DictOrNone = empty) -> DictOrNone:
        """
        获取展示/不展示的字段字典

        Args:
            projection: 展示/不展示的字段

        Returns:
            展示/不展示的字段字典

        Examples:
            >>> # `projection` 为 empty
            >>> CollectionSimpleUtils._get_projection()
            {
                "_id": 0,
            }

            >>> # `projection` 为 None
            >>> CollectionSimpleUtils._get_projection(projection=None)
            None

            >>> # `projection` 不为 None
            >>> CollectionSimpleUtils._get_projection(
            ...     projection={
            ...         "name": 1,
            ...         "age": 1,
            ...     },
            ... )
            {
                "_id": 0,
                "name": 1,
                "age": 1,
            }

            >>> # `projection` 不为 None，且需要展示 `_id` 字段
            >>> CollectionSimpleUtils._get_projection(
            ...     projection={
            ...         "_id": 1,
            ...         "name": 1,
            ...         "age": 1,
            ...     },
            ... )
            {
                "_id": 1,
                "name": 1,
                "age": 1,
            }
        """

        if projection is empty:
            return CollectionSimpleUtils.projection.copy()

        return projection

    @staticmethod
    def projection_shows(*args: str) -> DictStrInt:
        """
        需要展示的字段（默认不展示 `_id` 字段）

        Args:
            args: 需要展示的字段

        Returns:
            需要展示的字段-字典

        Examples:
            >>> # 需要展示的字段：`name`、`age`
            >>> CollectionSimpleUtils.projection_shows("name", "age")
            {
                "_id": 0,
                "name": 1,
                "age": 1,
            }

            >>> # 需要展示的字段：`_id`、`name`、`age`
            >>> CollectionSimpleUtils.projection_shows("_id", "name", "age")
            {
                "_id": 1,
                "name": 1,
                "age": 1,
            }
        """

        return {
            **CollectionSimpleUtils.projection,
            **{
                field: 1
                for field in args
            },
        }

    @staticmethod
    def projection_no_shows(*args: str) -> DictStrInt:
        """
        不需要展示的字段（默认不展示 _id 字段）

        Args:
            args: 不需要展示的字段

        Returns:
            不需要展示的字段-字典

        Examples:
            >>> # 不需要展示的字段：`name`、`age`
            >>> CollectionSimpleUtils.projection_no_shows("name", "age")
            {
                "_id": 0,
                "name": 0,
                "age": 0,
            }
        """

        return {
            **CollectionSimpleUtils.projection,
            **{
                field: 0
                for field in args
            },
        }


class CollectionBase(CollectionSimpleUtils):
    """
    MongoDB 集合 基类
    """

    collection_name: str = ""
    """ 集合名称 """

    def __init__(self) -> None:
        self.collection: Collection = mongo_db[self.collection_name]
        """ 集合 """

    def insert_one(
        self,
        document: dict,
        *args: Any,
        **kwargs: Any,
    ) -> tuple[bool, str]:
        """
        新增 单个 文档

        Args:
            document: 文档

        Returns:
            tuple[acknowledged, inserted_id]

        Examples:
            >>> # 新增单个文档（自动补充字段：`is_deleted`、`create_time`、`update_time`）
            >>> self.insert_one(
            ...     document={"name": "张三", "age": 18},
            ... )
            (
                True,                       # acknowledged  是否执行成功
                "645c5703ad048d260020ea78", # inserted_id   插入文档的 _id
            )

            >>> # 新增后，数据库中的数据：
                {
                    "_id" : ObjectId("645c5703ad048d260020ea78"),
                    "name" : "张三",
                    "age" : 18,
                    "is_deleted" : 0,
                    "create_time" : "2023-05-11 10:46:27",
                    "update_time" : "2023-05-11 10:46:27"
                }
        """

        _document: dict = {
            **document,
            **self.undeleted_mark,
            **self.get_time_dict("create_time", "update_time"),
        }

        _res: InsertOneResult = self.collection.insert_one(_document, *args, **kwargs)

        return (
            _res.acknowledged,
            str(_res.inserted_id),
        )

    def insert_many(
        self,
        documents: ListDict,
        *args: Any,
        **kwargs: Any,
    ) -> tuple[bool, ListStr]:
        """
        新增 多个 文档

        Args:
            documents: 文档列表

        Returns:
            tuple[acknowledged, inserted_ids]

        Examples:
            >>> # 新建多个文档（自动补充字段：`is_deleted`、`create_time`、`update_time`）
            >>> self.insert_many(
            ...     documents=[
            ...         {"name": "张三", "age": 18},
            ...         {"name": "李四", "age": 22},
            ...         {"name": "王五", "age": 19},
            ...     ],
            ... )
            (
                True,                           # acknowledged  是否执行成功
                [                               # inserted_ids  插入文档的 _id 列表
                    "645c588c0af90d9f1d8206b3",
                    "645c588c0af90d9f1d8206b4",
                    "645c588c0af90d9f1d8206b5",
                ],
            )

            >>> # 新建后，数据库中的数据：
                {
                    "_id" : ObjectId("645c588c0af90d9f1d8206b3"),
                    "name" : "张三",
                    "age" : 18,
                    "is_deleted" : 0,
                    "create_time" : "2023-05-11 10:53:00",
                    "update_time" : "2023-05-11 10:53:00"
                }
                {
                    "_id" : ObjectId("645c588c0af90d9f1d8206b4"),
                    "name" : "李四",
                    "age" : 22,
                    "is_deleted" : 0,
                    "create_time" : "2023-05-11 10:53:00",
                    "update_time" : "2023-05-11 10:53:00"
                }
                {
                    "_id" : ObjectId("645c588c0af90d9f1d8206b5"),
                    "name" : "王五",
                    "age" : 19,
                    "is_deleted" : 0,
                    "create_time" : "2023-05-11 10:53:00",
                    "update_time" : "2023-05-11 10:53:00"
                }
        """

        _documents: ListDict = [
            {
                **document,
                **self.undeleted_mark,
                **self.get_time_dict("create_time", "update_time"),
            }
            for document in documents
        ]

        _res: InsertManyResult = self.collection.insert_many(_documents, *args, **kwargs)

        return (
            _res.acknowledged,
            list(map(str, _res.inserted_ids)),
        )

    def replace_one(
        self,
        filter: dict,  # noqa: BuiltinShadowName
        replacement: dict,
        upsert: bool = False,
        filter_all: bool = False,
        *args: Any,
        **kwargs: Any,
    ) -> tuple[bool, int, int]:
        """
        替换 单个 文档

        Args:
            filter: 过滤条件
            replacement: 更新的文档
            upsert: 更新插入（存在：更新 / 不存在：创建）
            filter_all: 是否查询全部数据（包括已删除的数据）

        Returns:
            tuple[acknowledged, matched_count, modified_count]

        Examples:
            >>> # 数据库中的数据：
                {
                    "_id" : ObjectId("645c588c0af90d9f1d8206b3"),
                    "name" : "张三",
                    "age" : 18,
                    "is_deleted" : 0,
                    "create_time" : "2023-05-11 10:53:00",
                    "update_time" : "2023-05-11 10:53:00"
                }

            ---------------------------------------------------------------------

            >>> # 查询 {"name": "张三", "age": 18}，替换成 {"name": "李四"}
            >>> self.replace_one(
            ...     filter={"name": "张三", "age": 18},
            ...     replacement={"name": "李四"},
            ... )
            (
                True,   # acknowledged      是否执行成功
                1,      # matched_count     匹配到的文档数
                1,      # modified_count    修改的文档数
            )

            >>> # 替换后，数据库中的数据：
                {
                    "_id" : ObjectId("645c588c0af90d9f1d8206b3"),
                    "name" : "李四"
                }
        """

        _filter: dict = {
            **filter,
            **self.get_deleted_mark_dict(filter_all)
        }

        _res: UpdateResult = self.collection.replace_one(_filter, replacement, upsert, *args, **kwargs)

        return (
            _res.acknowledged,
            _res.matched_count,
            _res.modified_count,
        )

    def update_one(
        self,
        filter: dict,  # noqa: BuiltinShadowName
        update: dict,
        upsert: bool = False,
        filter_all: bool = False,
        *args: Any,
        **kwargs: Any,
    ) -> tuple[bool, int, int, StrOrNone]:
        """
        更新 单个 文档

        Args:
            filter: 过滤条件
            update: 更新的文档
            upsert: 更新插入（存在：更新 / 不存在：创建）
            filter_all: 是否查询全部数据（包括已删除的数据）

        Returns:
            tuple[acknowledged, matched_count, modified_count, upserted_id]

        Examples:
            >>> # 更新前，数据库中的数据：
                {
                    "_id" : ObjectId("645c588c0af90d9f1d8206b3"),
                    "name" : "张三",
                    "age" : 18,
                    "is_deleted" : 0,
                    "create_time" : "2023-05-11 10:53:00",
                    "update_time" : "2023-05-11 10:53:00"
                }

            >>> # 查询 {"name": "张三", "age": 18}，将 `age` 更新成 10（只更新 1 个文档）
            >>> self.update_one(
            ...     filter={"name": "张三", "age": 18},
            ...     update={"age": 10},
            ... )
            (
                True,   # acknowledged      是否执行成功
                1,      # matched_count     匹配到的文档数
                1,      # modified_count    修改的文档数
                None,   # upserted_id       插入的新文档的 _id
            )

            >>> # 更新后，数据库中的数据（`age`: 18 -> 10，`update_time` 更新为当前时间）：
                {
                    "_id" : ObjectId("645c588c0af90d9f1d8206b3"),
                    "name" : "张三",
                    "age" : 10,
                    "is_deleted" : 0,
                    "create_time" : "2023-05-11 10:53:00",
                    "update_time" : "2023-05-11 10:55:20"
                }

            ---------------------------------------------------------------------

            >>> # 查询 {"name": "张三", "age": 18}，将 `age` 更新成 10（只更新 1 个文档）
            >>> # 更新前，数据库中无任何数据：
            >>> self.update_one(
            ...     filter={"name": "张三", "age": 18},
            ...     update={"age": 10},
            ...     upsert=True,
            ... )
            (
                True,                       # acknowledged      是否执行成功
                0,                          # matched_count     匹配到的文档数
                0,                          # modified_count    修改的文档数
                "645c9a6034c6181c2ab79c30", # upserted_id       插入的新文档的 _id
            )

            >>> # 更新后，数据库中的数据（相当于插入了 1 条新的数据）：
                {
                    "_id" : ObjectId("645c9a6034c6181c2ab79c30"),
                    "age" : 10,
                    "name" : "张三",
                    "update_time" : "2023-05-11 15:33:52"
                }
        """

        _filter: dict = {
            **filter,
            **self.get_deleted_mark_dict(filter_all)
        }
        _update: dict = {
            "$set": {
                **update,
                **self.get_time_dict("update_time"),
            },
        }

        _res: UpdateResult = self.collection.update_one(_filter, _update, upsert, *args, **kwargs)

        return (
            _res.acknowledged,
            _res.matched_count,
            _res.modified_count,
            (
                str(_res.upserted_id)
                if _res.upserted_id is not None
                else _res.upserted_id
            ),
        )

    def update_many(
        self,
        filter: dict,  # noqa: BuiltinShadowName
        update: dict,
        upsert: bool = False,
        filter_all: bool = False,
        *args: Any,
        **kwargs: Any,
    ) -> tuple[bool, int, int, StrOrNone]:
        """
        更新 多个 文档

        Args:
            filter: 过滤条件
            update: 更新的文档
            upsert: 更新插入（存在：更新 / 不存在：创建）
            filter_all: 是否查询全部数据（包括已删除的数据）

        Returns:
            tuple[acknowledged, matched_count, modified_count, upserted_id]

        Examples:
            >>> # 数据库中的数据：
                {
                    "_id" : ObjectId("645c588c0af90d9f1d8206b3"),
                    "name" : "张三",
                    "age" : 18,
                    "is_deleted" : 0,
                    "create_time" : "2023-05-11 10:53:00",
                    "update_time" : "2023-05-11 10:53:00"
                }
                {
                    "_id" : ObjectId("645c588c0af90d9f1d8206b6"),
                    "name" : "赵六",
                    "age" : 18,
                    "is_deleted" : 0,
                    "create_time" : "2023-05-11 10:53:00",
                    "update_time" : "2023-05-11 10:53:00"
                }

            ---------------------------------------------------------------------

            >>> # 查询 {"age": 18}，将 `age` 更新成 20（更新所有匹配到的文档）
            >>> self.update_many(
            ...     filter={"age": 18},
            ...     update={"age": 20},
            ... )
            (
                True,   # acknowledged      是否执行成功
                2,      # matched_count     匹配到的文档数
                2,      # modified_count    修改的文档数
                None,   # upserted_id       插入的新文档的 _id
            )

            >>> # 更新后，数据库中的数据（`age`: 18 -> 20，`update_time` 更新为当前时间）：
                {
                    "_id" : ObjectId("645c588c0af90d9f1d8206b3"),
                    "name" : "张三",
                    "age" : 20,
                    "is_deleted" : 0,
                    "create_time" : "2023-05-11 10:53:00",
                    "update_time" : "2023-05-11 10:58:46"
                }
                {
                    "_id" : ObjectId("645c588c0af90d9f1d8206b6"),
                    "name" : "赵六",
                    "age" : 20,
                    "is_deleted" : 0,
                    "create_time" : "2023-05-11 10:53:00",
                    "update_time" : "2023-05-11 10:58:46"
                }
        """

        _filter: dict = {
            **filter,
            **self.get_deleted_mark_dict(filter_all)
        }
        _update: dict = {
            "$set": {
                **update,
                **self.get_time_dict("update_time"),
            },
        }

        _res: UpdateResult = self.collection.update_many(_filter, _update, upsert, *args, **kwargs)

        return (
            _res.acknowledged,
            _res.matched_count,
            _res.modified_count,
            (
                str(_res.upserted_id)
                if _res.upserted_id is not None
                else _res.upserted_id
            ),
        )

    def delete_one(
        self,
        filter: dict,  # noqa: BuiltinShadowName
        filter_all: bool = False,
        forced: bool = False,
        *args: Any,
        **kwargs: Any,
    ) -> tuple[bool, int] | tuple[bool, int, int, StrOrNone]:
        """
        删除 单个 文档（默认软删除）

        Args:
            filter: 过滤条件
            filter_all: 是否查询全部数据（包括已删除的数据）
            forced: 是否强制删除

        Returns:
            Union[
                tuple[acknowledged, deleted_count],
                tuple[acknowledged, matched_count, modified_count, upserted_id],
            ]

        Examples:
            >>> # 数据库中的数据（只有 1 条）：
                {
                    "_id" : ObjectId("645c588c0af90d9f1d8206b3"),
                    "name" : "张三",
                    "age" : 18,
                    "is_deleted" : 0,
                    "create_time" : "2023-05-11 10:53:00",
                    "update_time" : "2023-05-11 10:53:00"
                }

            ---------------------------------------------------------------------

            >>> # 查询 {"age": 18}，删除 1 条数据（软删除）
            >>> self.delete_one(
            ...     filter={"age": 18},
            ... )
            (
                True,   # acknowledged      是否执行成功
                1,      # matched_count     匹配到的文档数
                1,      # modified_count    修改的文档数
                None,   # upserted_id       插入的新文档的 _id
            )

            >>> # 删除后，数据库中的数据（还是 1 条）（`is_deleted`:0 -> 1，`update_time` 更新为当前时间）：
                {
                    "_id" : ObjectId("645c588c0af90d9f1d8206b3"),
                    "name" : "张三",
                    "age" : 18,
                    "is_deleted" : 1,
                    "create_time" : "2023-05-11 10:53:00",
                    "update_time" : "2023-05-11 16:18:03"
                }

            ---------------------------------------------------------------------

            >>> # 查询 {"age": 18}，删除 1 条数据（硬删除）
            >>> self.delete_one(
            ...     filter={"age": 18},
            ...     forced=True,
            ... )
            (
                True,   # acknowledged      是否执行成功
                1,      # deleted_count     删除的文档数
            )

            >>> # 删除后，数据库无数据
        """

        _filter: dict = {
            **filter,
            **self.get_deleted_mark_dict(filter_all)
        }

        if forced:
            _res: DeleteResult = self.collection.delete_one(_filter, *args, **kwargs)

            return (
                _res.acknowledged,
                _res.deleted_count,
            )

        return self.update_one(_filter, self.deleted_mark)

    def delete_many(
        self,
        filter: dict,  # noqa: BuiltinShadowName
        filter_all: bool = False,
        forced: bool = False,
        *args: Any,
        **kwargs: Any,
    ) -> tuple[bool, int] | tuple[bool, int, int, StrOrNone]:
        """
        删除 多个 文档（默认软删除）

        Args:
            filter: 过滤条件
            filter_all: 是否查询全部数据（包括已删除的数据）
            forced: 是否强制删除

        Returns:
            Union[
                tuple[acknowledged, deleted_count],
                tuple[acknowledged, matched_count, modified_count, upserted_id],
            ]

        Examples:
            >>> # 删除前，数据库中的数据：
                {
                    "_id" : ObjectId("645c588c0af90d9f1d8206b3"),
                    "name" : "张三",
                    "age" : 18,
                    "is_deleted" : 0,
                    "create_time" : "2023-05-11 10:53:00",
                    "update_time" : "2023-05-11 10:53:00"
                }
                {
                    "_id" : ObjectId("645c588c0af90d9f1d8206b6"),
                    "name" : "赵六",
                    "age" : 18,
                    "is_deleted" : 0,
                    "create_time" : "2023-05-11 10:53:00",
                    "update_time" : "2023-05-11 10:53:00"
                }

            ---------------------------------------------------------------------

            >>> # 查询 {"age": 18}，删除匹配到的所有数据（软删除）
            >>> self.delete_many(
            ...     filter={"age": 18},
            ... )
            (
                True,   # acknowledged      是否执行成功
                2,      # matched_count     匹配到的文档数
                2,      # modified_count    修改的文档数
                None,   # upserted_id       插入的新文档的 _id
            )

            >>> # 删除后，数据库中的数据（还是 2 条）（`is_deleted`:0 -> 1，`update_time` 更新为当前时间）：
                {
                    "_id" : ObjectId("645c588c0af90d9f1d8206b3"),
                    "name" : "张三",
                    "age" : 18,
                    "is_deleted" : 1,
                    "create_time" : "2023-05-11 10:53:00",
                    "update_time" : "2023-05-11 10:57:00"
                }
                {
                    "_id" : ObjectId("645c588c0af90d9f1d8206b6"),
                    "name" : "赵六",
                    "age" : 18,
                    "is_deleted" : 1,
                    "create_time" : "2023-05-11 10:53:00",
                    "update_time" : "2023-05-11 10:57:00"
                }

            ---------------------------------------------------------------------

            >>> # 查询（包括已删除） {"age": 18}，删除匹配到的所有数据
            >>> self.delete_many(
            ...     filter={"age": 18},
            ...     filter_all=True,
            ...     forced=True,
            ... )
            (
                True,   # acknowledged      是否执行成功
                2,      # deleted_count     删除的文档数
            )

            >>> # 删除后，数据库无数据
        """

        _filter: dict = {
            **filter,
            **self.get_deleted_mark_dict(filter_all)
        }

        if forced:
            _res: DeleteResult = self.collection.delete_many(_filter, *args, **kwargs)

            return (
                _res.acknowledged,
                _res.deleted_count,
            )

        return self.update_many(_filter, self.deleted_mark)

    def find_one(
        self,
        filter: DictOrNone = None,  # noqa: BuiltinShadowName
        projection: DictStrInt | None = empty,
        filter_all: bool = False,
        *args: Any,
        **kwargs: Any,
    ) -> dict:
        """
        查询 单个 文档（默认不返回 _id）

        Args:
            filter: 过滤条件
            projection: 展示/不展示的字段字典
            filter_all: 是否查询全部数据（包括已删除的数据）

        Returns:
            查询到的文档（单个）

        Examples:
            >>> # 数据库中的数据：
                {
                    "_id" : ObjectId("645c588c0af90d9f1d8206b3"),
                    "name" : "张三",
                    "age" : 18,
                    "is_deleted" : 0,
                    "create_time" : "2023-05-11 10:53:00",
                    "update_time" : "2023-05-11 10:53:00"
                }
                {
                    "_id" : ObjectId("645c588c0af90d9f1d8206b6"),
                    "name" : "赵六",
                    "age" : 18,
                    "is_deleted" : 0,
                    "create_time" : "2023-05-11 10:53:00",
                    "update_time" : "2023-05-11 10:53:00"
                }

            ---------------------------------------------------------------------

            >>> # 查询 {"name": "张三"}，返回所有字段（不包括 _id）
            >>> self.find_one(
            ...     filter={"name": "张三"},
            ... )
            {
                "_id" : ObjectId("645c588c0af90d9f1d8206b3"),
                "name" : "张三",
                "age" : 18,
                "is_deleted" : 0,
                "create_time" : "2023-05-11 10:53:00",
                "update_time" : "2023-05-11 10:53:00"
            }

            ---------------------------------------------------------------------

            >>> # 查询 {"name": "张三"}，返回所有字段（包括 _id）
            >>> self.find_one(
            ...     filter={"name": "张三"},
            ...     projection=None,
            ... )
            {
                "name": "张三",
                "age": 18,
                "is_deleted": 0,
                "create_time": "2023-05-11 10:53:00",
                "update_time": "2023-05-11 16:18:03"
            }

            ---------------------------------------------------------------------

            >>> # 查询 {"name": "张三"}，返回字段：`name`、`age`
            >>> self.find_one(
            ...     filter={"name": "张三"},
            ...     projection=self.projection_shows("name", "age"),
            ... )
            {
                "name": "张三",
                "age": 18,
            }
        """

        _filter: dict = {
            **(
                {}
                if filter is None
                else filter
            ),
            **self.get_deleted_mark_dict(filter_all)
        }

        _projection: DictStrInt | None = (
            self.projection
            if projection is empty
            else projection
        )

        return self.collection.find_one(_filter, _projection, *args, **kwargs) or {}

    def find(
        self,
        filter: DictOrNone = None,  # noqa: BuiltinShadowName
        projection: DictStrInt | None = empty,
        filter_all: bool = False,
        *args: Any,
        **kwargs: Any,
    ) -> ListDict:
        """
        查询 多个 文档（默认不返回 _id）

        Args:
            filter: 过滤条件
            projection: 展示/不展示的字段字典
            filter_all: 是否查询全部数据（包括已删除的数据）

        Returns:
            查询到的文档（多个）

        Examples:
            >>> # 数据库中的数据：
                {
                    "_id" : ObjectId("645c588c0af90d9f1d8206b3"),
                    "name" : "张三",
                    "age" : 18,
                    "is_deleted" : 0,
                    "create_time" : "2023-05-11 10:53:00",
                    "update_time" : "2023-05-11 10:53:00"
                }
                {
                    "_id" : ObjectId("645c588c0af90d9f1d8206b6"),
                    "name" : "赵六",
                    "age" : 18,
                    "is_deleted" : 1,
                    "create_time" : "2023-05-11 10:53:00",
                    "update_time" : "2023-05-11 10:53:00"
                }

            ---------------------------------------------------------------------

            >>> # 查询所有数据（不包括已删除），返回所有字段（不包括 `_id`）
            >>> self.find()
            {
                "name": "张三",
                "age": 18,
                "is_deleted": 0,
                "create_time": "2023-05-11 10:53:00",
                "update_time": "2023-05-11 16:18:03"
            }

            ---------------------------------------------------------------------

            >>> # 查询所有数据（包括已删除），返回所有字段（不包括 `_id`）
            >>> self.find(
            >>>     filter_all=True,
            >>> )
            [
                {
                    "name" : "张三",
                    "age" : 18,
                    "is_deleted" : 0,
                    "create_time" : "2023-05-11 10:53:00",
                    "update_time" : "2023-05-11 10:53:00"
                },
                {
                    "name" : "赵六",
                    "age" : 18,
                    "is_deleted" : 1,
                    "create_time" : "2023-05-11 10:53:00",
                    "update_time" : "2023-05-11 10:53:00"
                },
            ]
        """

        _filter: dict = {
            **(
                {}
                if filter is None
                else filter
            ),
            **self.get_deleted_mark_dict(filter_all)
        }

        _projection: DictStrInt | None = (
            self.projection
            if projection is empty
            else projection
        )

        return list(self.collection.find(_filter, _projection, *args, **kwargs))

    def count_documents(
        self,
        filter: DictOrNone = None,  # noqa: BuiltinShadowName
        filter_all: bool = False,
        *args: Any,
        **kwargs: Any,
    ) -> int:
        """
        计算该集合中的文档数

        Args:
            filter: 过滤条件
            filter_all: 是否查询全部数据（包括已删除的数据）

        Returns:
            文档总数

        Examples:
            >>> # 数据库中的数据：
                {
                    "_id" : ObjectId("645c588c0af90d9f1d8206b3"),
                    "name" : "张三",
                    "age" : 18,
                    "is_deleted" : 0,
                    "create_time" : "2023-05-11 10:53:00",
                    "update_time" : "2023-05-11 10:53:00"
                }
                {
                    "_id" : ObjectId("645c588c0af90d9f1d8206b6"),
                    "name" : "赵六",
                    "age" : 18,
                    "is_deleted" : 1,
                    "create_time" : "2023-05-11 10:53:00",
                    "update_time" : "2023-05-11 10:53:00"
                }

            ---------------------------------------------------------------------

            >>> # 计算所有数据条数（不包括已删除），
            >>> self.count_documents()
            1

            ---------------------------------------------------------------------

            >>> # 计算所有数据条数（包括已删除）
            >>> self.count_documents(filter_all=True)
            2
        """

        _filter: dict = {
            **(
                {}
                if filter is None
                else filter
            ),
            **self.get_deleted_mark_dict(filter_all)
        }

        return self.collection.count_documents(_filter, *args, **kwargs)

    def aggregate(
        self,
        pipeline: list[Mapping[str, Any]],
        *args: Any,
        **kwargs: Any,
    ) -> ListDict:
        """
        聚合

        Args:
            pipeline: 聚合管道

        Returns:
            查询到的文档
        """

        return list(self.collection.aggregate(pipeline, *args, **kwargs))

    def distinct(
        self,
        key: str,
        filter: DictOrNone = None,  # noqa: BuiltinShadowName
        filter_all: bool = False,
        *args: Any,
        **kwargs: Any,
    ) -> list:
        """
        获取去重后的字段值列表

        Args:
            key: 字段名
            filter: 过滤条件
            filter_all: 是否查询全部数据（包括已删除的数据）

        Returns:
            去重后的字段值列表

        Examples:
            >>> # 数据库中的数据：
                {
                    "_id" : ObjectId("645cb418c9bc1589fa48208a"),
                    "name" : "张三",
                    "age" : 18,
                    "gender" : "男",
                    "is_deleted" : 0,
                    "create_time" : "2023-05-11 17:23:36",
                    "update_time" : "2023-05-11 17:23:36"
                }
                {
                    "_id" : ObjectId("645cb418c9bc1589fa48208b"),
                    "name" : "李四",
                    "age" : 22,
                    "gender" : "女",
                    "is_deleted" : 0,
                    "create_time" : "2023-05-11 17:23:36",
                    "update_time" : "2023-05-11 17:23:36"
                }
                {
                    "_id" : ObjectId("645cb418c9bc1589fa48208c"),
                    "name" : "王五",
                    "age" : 19,
                    "gender" : "男",
                    "is_deleted" : 0,
                    "create_time" : "2023-05-11 17:23:36",
                    "update_time" : "2023-05-11 17:23:36"
                }
                {
                    "_id" : ObjectId("645cb418c9bc1589fa48208d"),
                    "name" : "赵六",
                    "age" : 19,
                    "gender" : "男",
                    "is_deleted" : 0,
                    "create_time" : "2023-05-11 17:23:36",
                    "update_time" : "2023-05-11 17:23:36"
                }
                {
                    "_id" : ObjectId("645cb418c9bc1589fa48208e"),
                    "name" : "孙七",
                    "age" : 18,
                    "gender" : "女",
                    "is_deleted" : 0,
                    "create_time" : "2023-05-11 17:23:36",
                    "update_time" : "2023-05-11 17:23:36"
                }
                {
                    "_id" : ObjectId("645cb418c9bc1589fa48208f"),
                    "name" : "周八",
                    "age" : 18,
                    "gender" : "男",
                    "is_deleted" : 0,
                    "create_time" : "2023-05-11 17:23:36",
                    "update_time" : "2023-05-11 17:23:36"
                }

            ---------------------------------------------------------------------

            >>> # 查询所有数据，对 `age` 字段去重
            >>> self.distinct("age")
            [18, 19, 22]

            ---------------------------------------------------------------------

            >>> # 查询 {"age": 18} 的数据，对 `age` 字段去重
            >>> self.distinct("gender", {"age": 18})
            ["女", "男"]
        """

        _filter: dict = {
            **(
                {}
                if filter is None
                else filter
            ),
            **self.get_deleted_mark_dict(filter_all)
        }

        return self.collection.distinct(key, _filter, *args, **kwargs)

    def bulk_write(
        self,
        requests: list[InsertOne | UpdateOne | UpdateMany | ReplaceOne | DeleteOne | DeleteMany],
        *args: Any,
        **kwargs: Any,
    ) -> BulkWriteResult:
        """
        批量写入

        Args:
            requests: 批量写入请求

        Returns:
            批量写入结果

                - `bulk_api_result` (dict): 批量写入的结果
                - `inserted_count` (int): 插入的文档数
                - `matched_count` (int): 匹配到的文档数
                - `modified_count` (int): 修改的文档数
                - `deleted_count` (int): 删除的文档数
                - `upserted_count` (int): 插入的新文档数
                - `upserted_ids` (DictOrNone): 插入的新文档的 _id 字典
        """

        return self.collection.bulk_write(requests, *args, **kwargs)

    def paginated_query(
        self,
        filter: DictOrNone = None,  # noqa: BuiltinShadowName
        projection: DictStrInt | None = empty,
        order: list[dict[str, IntOrStr]] | None = None,
        filter_all: bool = False,
        page: int = 1,
        page_size: int = 10,
        *args,
        **kwargs,
    ) -> dict[str, int | ListDict]:
        """
        分页查询

        Args:
            filter: 过滤条件
            projection: 展示/不展示的字段字典
            order: 排序字段
            filter_all: 是否查询全部数据（包括已删除的数据）
            page: 页码
            page_size: 每页数据条数

        Returns:
            分页后的数据

        Examples:
            >>> # 数据库中的数据：
                {
                    "_id" : ObjectId("645cb418c9bc1589fa48208a"),
                    "name" : "张三",
                    "age" : 18,
                    "gender" : "男",
                    "is_deleted" : 0,
                    "create_time" : "2023-05-11 17:23:36",
                    "update_time" : "2023-05-11 17:23:36"
                }
                {
                    "_id" : ObjectId("645cb418c9bc1589fa48208b"),
                    "name" : "李四",
                    "age" : 22,
                    "gender" : "女",
                    "is_deleted" : 0,
                    "create_time" : "2023-05-11 17:23:36",
                    "update_time" : "2023-05-11 17:23:36"
                }
                {
                    "_id" : ObjectId("645cb418c9bc1589fa48208c"),
                    "name" : "王五",
                    "age" : 19,
                    "gender" : "男",
                    "is_deleted" : 0,
                    "create_time" : "2023-05-11 17:23:36",
                    "update_time" : "2023-05-11 17:23:36"
                }
                {
                    "_id" : ObjectId("645cb418c9bc1589fa48208d"),
                    "name" : "赵六",
                    "age" : 19,
                    "gender" : "男",
                    "is_deleted" : 0,
                    "create_time" : "2023-05-11 17:23:36",
                    "update_time" : "2023-05-11 17:23:36"
                }
                {
                    "_id" : ObjectId("645cb418c9bc1589fa48208e"),
                    "name" : "孙七",
                    "age" : 18,
                    "gender" : "女",
                    "is_deleted" : 0,
                    "create_time" : "2023-05-11 17:23:36",
                    "update_time" : "2023-05-11 17:23:36"
                }
                {
                    "_id" : ObjectId("645cb418c9bc1589fa48208f"),
                    "name" : "周八",
                    "age" : 18,
                    "gender" : "男",
                    "is_deleted" : 0,
                    "create_time" : "2023-05-11 17:23:36",
                    "update_time" : "2023-05-11 17:23:36"
                }

            ---------------------------------------------------------------------

            >>> # 查询全部数据，每页 5 条，只展示：`name`、`age`、`gender` 字段
            >>> self.paginated_query(
            ...     projection=self.projection_shows("name", "age", "gender"),
            ...     page_size=5,
            ... )
            {
                "page": 1,          # 第几页
                "page_size": 5,     # 每页几条
                "page_count": 2,    # 总页数
                "data_count": 6,    # 总数据条数
                "data_list": [      # 分页后的数据
                    {
                        "name" : "张三",
                        "age" : 18,
                        "gender" : "男",
                    },
                    {
                        "name" : "李四",
                        "age" : 22,
                        "gender" : "女",
                    },
                    {
                        "name" : "王五",
                        "age" : 19,
                        "gender" : "男",
                    },
                    {
                        "name" : "赵六",
                        "age" : 19,
                        "gender" : "男",
                    },
                    {
                        "name" : "孙七",
                        "age" : 18,
                        "gender" : "女",
                    },
                ],
            }

            ---------------------------------------------------------------------

            >>> # 查询全部数据，每页 3 条，只展示：`name`、`age`、`gender` 字段，根据 `age` 降序、`gender` 升序
            >>> self.paginated_query(
            ...     projection=self.projection_shows("name", "age", "gender"),
            ...     page_size=3,
            ...     order=[
            ...         {
            ...             "field": "age",
            ...             "direction": -1,
            ...         },
            ...         {
            ...             "field": "name",
            ...             "direction": 1,
            ...         },
            ...     ],
            ... )
            {
                "page": 1,
                "page_size": 3,
                "page_count": 2,
                "data_count": 6,
                "data_list": [
                    {
                        "name": "李四",
                        "age": 22,
                        "gender": "女",
                    },
                    {
                        "name": "王五",
                        "age": 19,
                        "gender": "男",
                    },
                    {
                        "name": "赵六",
                        "age": 19,
                        "gender": "男",
                    },
                ],
            }
        """

        _filter: dict = {
            **(
                {}
                if filter is None
                else filter
            ),
            **self.get_deleted_mark_dict(filter_all)
        }

        _projection: DictStrInt | None = (
            self.projection
            if projection is empty
            else projection
        )

        data_count: int = self.count_documents(_filter)
        """ 总数据条数 """

        page_count: int = math.ceil(data_count / page_size)
        """ 总页数 """

        skip_count: int = page_size * (page - 1)
        """ 当前查询需要跳过的数据条数 """

        raw_data: Cursor = self.collection.find(_filter, _projection, *args, **kwargs)
        """ 查询的原始数据：Cursor 对象，类似于迭代器，只能循环 1 次 """

        # 排序
        if order is not None:
            raw_data = raw_data.sort(
                key_or_list=[
                    (
                        str(item["field"]),
                        int(item["direction"]),
                    )
                    for item in order
                ]
            )

        data_list: ListDict = list(raw_data.skip(skip_count).limit(page_size))
        """ 分页后的数据 """

        return {
            "page": page,  # ->             第几页
            "page_size": page_size,  # ->   每页几条
            "page_count": page_count,  # -> 总页数
            "data_count": data_count,  # -> 总数据条数
            "data_list": data_list,  # ->   分页后的数据
        }
